<?php
require_once __DIR__ . '/../../vendor/autoload.php';
//date_default_timezone_set("PRC");

$connection = new \PhpAmqpLib\Connection\AMQPStreamConnection('127.0.0.1', 5672, 'guest', 'guest');

$channel = $connection->channel();
// 定义队列
$channel->queue_declare('first', false, true,false, false);
$channel->basic_qos(0, 1, false);
var_dump('duile—server is  ready，');
$callback = function ($msg){

    // 接收到消息
    $content = json_decode($msg->body,true);

    // 对数据库操作
    $con = mysqli_connect('localhost','root','root','new_nydata');
//    $con = mysqli_connect('106.15.183.188','new_nydata','new_nydata','new_nydata');
    if(!$con){
        var_dump('mysql connect error');
    }

    // 把设备和传感器的mac_id和对应的自增id放入redis中
    $Redis = new Redis();
    $Redis->connect('127.0.0.1', 6379);
    var_dump("redis is ready");
    // 设备
    $dev_sql = "select id,dev_mac from nyd_dev";
    $dev_res = mysqli_query($con, $dev_sql);
    while ($row = $dev_res->fetch_assoc()) {
        if (!$Redis->get($row['dev_mac'])) {
            $Redis->set($row['dev_mac'], 'nyd_dev-dev_mac-'.$row['id']);
        }
    }

    // 传感器
    $sec_sql = "select id,sensor_mac  from nyd_sensor";
    $sec_res = mysqli_query($con, $sec_sql);
    while ($row = $sec_res->fetch_assoc()) {
        if (!$Redis->get($row['sensor_mac'])) {
            $Redis->set($row['sensor_mac'], 'nyd_sensor-sensor_mac-'.$row['id']);
        }
    }

    // 上传设备状态
    if($content['type'] == 2){
        $insert_data = "";
        $data = $content['data'];
        for ($i = 0; $i < count($data); $i++) {
            $value = $Redis->get($data[$i]['mac_id']);
            list($table,$field,$id) = explode('-',$value);
            // 状态更新
            $sql = "update ".$table." set is_bad =".$data[$i]['is_bad'].",is_open =".$data[$i]['is_open']." where ".$field." = '".$data[$i]['mac_id']."'";
//            var_dump('111111\n');
//            var_dump($sql);
            mysqli_query($con,$sql);

            // 状态插入
            if ($i != count($data) - 1) {
                // $insert_data .= "('" . $data[$i]['mac_id'] . "','" . $content['gateway_id'] . "','" . date("Y-m-d H:i:s",$content['time']) . "'," . $data[$i]['is_open'] . "," . $data[$i]['is_bad'] . "),";
                $insert_data .= "('" . $data[$i]['mac_id'] . "','" . $content['gateway_id'] . "','" . date("Y-m-d H:i:s",time()) . "'," . $data[$i]['is_open'] . "," . $data[$i]['is_bad'] . "),";
            } else {
                // $insert_data .= "('" . $data[$i]['mac_id'] . "','" . $content['gateway_id'] . "','" . date("Y-m-d H:i:s",$content['time']) . "'," . $data[$i]['is_open'] . "," . $data[$i]['is_bad'] . ")";
                $insert_data .= "('" . $data[$i]['mac_id'] . "','" . $content['gateway_id'] . "','" . date("Y-m-d H:i:s",time()) . "'," . $data[$i]['is_open'] . "," . $data[$i]['is_bad'] . ")";
            }
        }

        $sql = "insert into nyd_dev_status (mac_id,gateway_id,now_time,is_open,is_bad) VALUES " . $insert_data;
//        var_dump($sql);
//        var_dump('111111\n');

    } // 上传汇总数据
    elseif ($content['type'] == 3) {

        // 剥离数据 分别入库
        $insert_data = "";
        $data = $content['data'];

        for ($i = 0; $i < count($data); $i++) {
            $value = $Redis->get($data[$i]['mac_id']);
            list($table,$field,$id) = explode('-',$value);
            if ($i != count($data) - 1) {
                // $insert_data .= "(" . $id . ",'" . (string)$data[$i]['data'] . "'," . $data[$i]['is_open'] . ",'" . date("Y-m-d H:i:s",$content['time']) . "','" . $content['gateway_id'] . "'),";
                $insert_data .= "(" . $id . ",'" . (string)$data[$i]['data'] . "'," . $data[$i]['is_open'] . ",'" . date("Y-m-d H:i:s",time()) . "','" . $content['gateway_id'] . "'),";
            } else {
                // $insert_data .= "(" . $id . ",'" . (string)$data[$i]['data'] . "'," . $data[$i]['is_open'] . ",'" . date("Y-m-d H:i:s",$content['time']) . "','" . $content['gateway_id'] . "')";
                $insert_data .= "(" . $id . ",'" . (string)$data[$i]['data'] . "'," . $data[$i]['is_open'] . ",'" . date("Y-m-d H:i:s",time()) . "','" . $content['gateway_id'] . "')";
            }
        }

        $sql = "insert into nyd_data (dev_id,data,is_open,now_time,gateway_id) VALUES " . $insert_data;
    }

    if(mysqli_query($con,$sql)){
        var_dump('success insert to mysql');
    }else{
        var_dump('Error         '.$sql);
    }
    $con->close();

    // 接收到消息后确认
    $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);

};

$channel->basic_consume('first', '', false, false, false, false,$callback);
while(count($channel->callbacks)) {
    $channel->wait();
}
//function send_heartbeat($connection)
//{
//    $connection->getIO()->read(0);
//}
$channel->close();

//function read($len){
//    if (is_null($this->sock)) {
//        throw new AMQPSocketException(sprintf(
//            'Socket was null! Last SocketError was: %s',
//            socket_strerror(socket_last_error())
//        ));
//    }
//
//    $this->check_heartbeat();
//
//    list($timeout_sec, $timeout_uSec) = MiscHelper::splitSecondsMicroseconds($this->read_timeout);
//    $read_start = microtime(true);
//    $read = 0;
//    $data = '';
//    while ($read < $len) {
//        $buffer = null;
//        $result = socket_recv($this->sock, $buffer, $len - $read, 0);
//        if ($result === 0) {
//            // From linux recv() manual:
//            // When a stream socket peer has performed an orderly shutdown,
//            // the return value will be 0 (the traditional "end-of-file" return).
//            // http://php.net/manual/en/function.socket-recv.php#47182
//            $this->close();
//            throw new AMQPConnectionClosedException('Broken pipe or closed connection');
//        }
//
//        if (empty($buffer)) {
//            $read_now = microtime(true);
//            $t_read = $read_now - $read_start;
//            if ($t_read > $this->read_timeout) {
//                throw new AMQPTimeoutException('Too many read attempts detected in SocketIO');
//            }
//            $this->select($timeout_sec, $timeout_uSec);
//            continue;
//        }
//
//        $read += mb_strlen($buffer, 'ASCII');
//        $data .= $buffer;
//    }
//
//    if (mb_strlen($data, 'ASCII') != $len) {
//        throw new AMQPIOException(sprintf(
//            'Error reading data. Received %s instead of expected %s bytes',
//            mb_strlen($data, 'ASCII'),
//            $len
//        ));
//    }
//
//    $this->last_read = microtime(true);
//}
$connection->close();
